package drds.binlog.metadata;

import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.Position;
import drds.binlog.common.position.PositionRange;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.common.zookeeper.Zookeepers;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
 * zk 版本的 canal manager， 存储结构：
 * <p>
 * <pre>
 * /otter
 *    canal
 *      destinationToClientInfoListMap
 *        dest1
 *          client1
 *            filter
 *            batch_mark
 *              1
 *              2
 *              3
 * </pre>
 */
public class ZooKeeperMetaDataManager extends AbstractLifeCycle implements MetaDataManager
{

    private static final String ENCODE = "UTF-8";
    private ZooKeeperClient zooKeeperClient;

    public void start()
    {
        super.start();

        Assert.notNull(zooKeeperClient);
    }

    public void stop()
    {
        super.stop();
    }

    public void subscribe(ClientInfo clientInfo) throws MetaDataManagerException
    {
        String path = Zookeepers.getClientIdNodePath(clientInfo.getDestination(),
                clientInfo.getClientId());

        try
        {
            zooKeeperClient.createPersistent(path, true);
        } catch (ZkNodeExistsException e)
        {
            // ignore
        }
        if (clientInfo.hasFilter())
        {
            String filterPath = Zookeepers.getFilterPath(clientInfo.getDestination(),
                    clientInfo.getClientId());

            byte[] bytes = null;
            try
            {
                bytes = clientInfo.getFilter().getBytes(ENCODE);
            } catch (UnsupportedEncodingException e)
            {
                throw new MetaDataManagerException(e);
            }

            try
            {
                zooKeeperClient.createPersistent(filterPath, bytes);
            } catch (ZkNodeExistsException e)
            {
                // ignore
                zooKeeperClient.writeData(filterPath, bytes);
            }
        }
    }

    public boolean hasSubscribed(ClientInfo clientInfo) throws MetaDataManagerException
    {
        String path = Zookeepers.getClientIdNodePath(clientInfo.getDestination(),
                clientInfo.getClientId());
        return zooKeeperClient.exists(path);
    }

    public void unsubscribe(ClientInfo clientInfo) throws MetaDataManagerException
    {
        String path = Zookeepers.getClientIdNodePath(clientInfo.getDestination(),
                clientInfo.getClientId());
        zooKeeperClient.deleteRecursive(path); // 递归删除所有信息
    }

    public List<ClientInfo> getClientInfoList(String destination) throws MetaDataManagerException
    {
        String path = Zookeepers.getDestinationPath(destination);
        List<String> childs = null;
        try
        {
            childs = zooKeeperClient.getChildren(path);
        } catch (ZkNoNodeException e)
        {
            // ignore
        }

        if (CollectionUtils.isEmpty(childs))
        {
            return new ArrayList<ClientInfo>();
        }
        List<Short> clientIds = new ArrayList<Short>();
        for (String child : childs)
        {
            if (StringUtils.isNumeric(child))
            {
                clientIds.add(Zookeepers.getClientId(child));
            }
        }

        Collections.sort(clientIds); // 进行一个排序
        List<ClientInfo> clientIdentities = Lists.newArrayList();
        for (Short clientId : clientIds)
        {
            path = Zookeepers.getFilterPath(destination, clientId);
            byte[] bytes = zooKeeperClient.readData(path, true);
            String filter = null;
            if (bytes != null)
            {
                try
                {
                    filter = new String(bytes, ENCODE);
                } catch (UnsupportedEncodingException e)
                {
                    throw new MetaDataManagerException(e);
                }
            }
            clientIdentities.add(new ClientInfo(destination, clientId, filter));
        }

        return clientIdentities;
    }

    public Position getPosition(ClientInfo clientInfo) throws MetaDataManagerException
    {
        String path = Zookeepers.getCursorPath(clientInfo.getDestination(), clientInfo.getClientId());

        byte[] data = zooKeeperClient.readData(path, true);
        if (data == null || data.length == 0)
        {
            return null;
        }

        return JsonUtils.unmarshalFromByte(data, Position.class);
    }

    public void updatePosition(ClientInfo clientInfo, Position position) throws MetaDataManagerException
    {
        String path = Zookeepers.getCursorPath(clientInfo.getDestination(), clientInfo.getClientId());
        byte[] data = JsonUtils.marshalToByte(position, SerializerFeature.WriteClassName);
        try
        {
            zooKeeperClient.writeData(path, data);
        } catch (ZkNoNodeException e)
        {
            zooKeeperClient.createPersistent(path, data, true);// 第一次节点不存在，则尝试重建
        }
    }

    public Long addPositionRange(ClientInfo clientInfo, PositionRange positionRange) throws MetaDataManagerException
    {
        String path = Zookeepers.getBatchMarkPath(clientInfo.getDestination(), clientInfo.getClientId());
        byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
        String batchPath = zooKeeperClient.createPersistentSequential(path + Zookeepers.ZOOKEEPER_SEPARATOR,
                data,
                true);
        String batchIdString = StringUtils.substringAfterLast(batchPath, Zookeepers.ZOOKEEPER_SEPARATOR);
        return Zookeepers.getBatchMarkId(batchIdString);
    }

    public void addPositionRange(ClientInfo clientInfo, Long batchId, PositionRange positionRange)
            throws MetaDataManagerException
    {
        String path = Zookeepers.getBatchMarkWithIdPath(clientInfo.getDestination(),
                clientInfo.getClientId(),
                batchId);
        byte[] data = JsonUtils.marshalToByte(positionRange, SerializerFeature.WriteClassName);
        zooKeeperClient.createPersistent(path, data, true);
    }

    public PositionRange removePositionRange(ClientInfo clientInfo, Long batchId) throws MetaDataManagerException
    {
        String batchsPath = Zookeepers.getBatchMarkPath(clientInfo.getDestination(),
                clientInfo.getClientId());
        List<String> nodes = zooKeeperClient.getChildren(batchsPath);
        if (CollectionUtils.isEmpty(nodes))
        {
            // 没有batch记录
            return null;
        }

        // 找到最小的Id
        ArrayList<Long> batchIds = new ArrayList<Long>(nodes.size());
        for (String batchIdString : nodes)
        {
            batchIds.add(Long.valueOf(batchIdString));
        }
        Long minBatchId = Collections.min(batchIds);
        if (!minBatchId.equals(batchId))
        {
            // 检查一下提交的ack/rollback，必须按batchId分出去的顺序提交，否则容易出现丢数据
            throw new MetaDataManagerException(String.format("batchId:%d is not the firstly:%d", batchId, minBatchId));
        }

        if (!batchIds.contains(batchId))
        {
            // 不存在对应的batchId
            return null;
        }
        PositionRange positionRange = getPositionRange(clientInfo, batchId);
        if (positionRange != null)
        {
            String path = Zookeepers.getBatchMarkWithIdPath(clientInfo.getDestination(),
                    clientInfo.getClientId(),
                    batchId);
            zooKeeperClient.delete(path);
        }

        return positionRange;
    }

    public PositionRange getPositionRange(ClientInfo clientInfo, Long batchId) throws MetaDataManagerException
    {
        String path = Zookeepers.getBatchMarkWithIdPath(clientInfo.getDestination(),
                clientInfo.getClientId(),
                batchId);
        byte[] data = zooKeeperClient.readData(path, true);
        if (data == null)
        {
            return null;
        }

        PositionRange positionRange = JsonUtils.unmarshalFromByte(data, PositionRange.class);
        return positionRange;
    }

    public void clearBatchIdToPositionRangeMap(ClientInfo clientInfo) throws MetaDataManagerException
    {
        String path = Zookeepers.getBatchMarkPath(clientInfo.getDestination(), clientInfo.getClientId());
        List<String> batchChilds = zooKeeperClient.getChildren(path);

        for (String batchChild : batchChilds)
        {
            String batchPath = path + Zookeepers.ZOOKEEPER_SEPARATOR + batchChild;
            zooKeeperClient.delete(batchPath);
        }
    }

    public PositionRange getLastestPositionRange(ClientInfo clientInfo)
    {
        String path = Zookeepers.getBatchMarkPath(clientInfo.getDestination(), clientInfo.getClientId());
        List<String> nodes = null;
        try
        {
            nodes = zooKeeperClient.getChildren(path);
        } catch (ZkNoNodeException e)
        {
            // ignore
        }

        if (CollectionUtils.isEmpty(nodes))
        {
            return null;
        }
        // 找到最大的Id
        ArrayList<Long> batchIds = new ArrayList<Long>(nodes.size());
        for (String batchIdString : nodes)
        {
            batchIds.add(Long.valueOf(batchIdString));
        }
        Long maxBatchId = Collections.max(batchIds);
        PositionRange result = getPositionRange(clientInfo, maxBatchId);
        if (result == null)
        { // 出现为null，说明zk节点有变化，重新获取
            return getLastestPositionRange(clientInfo);
        } else
        {
            return result;
        }
    }

    public PositionRange getFirstPositionRange(ClientInfo clientInfo)
    {
        String path = Zookeepers.getBatchMarkPath(clientInfo.getDestination(), clientInfo.getClientId());
        List<String> nodes = null;
        try
        {
            nodes = zooKeeperClient.getChildren(path);
        } catch (ZkNoNodeException e)
        {
            // ignore
        }

        if (CollectionUtils.isEmpty(nodes))
        {
            return null;
        }
        // 找到最小的Id
        ArrayList<Long> batchIds = new ArrayList<Long>(nodes.size());
        for (String batchIdString : nodes)
        {
            batchIds.add(Long.valueOf(batchIdString));
        }
        Long minBatchId = Collections.min(batchIds);
        PositionRange result = getPositionRange(clientInfo, minBatchId);
        if (result == null)
        { // 出现为null，说明zk节点有变化，重新获取
            return getFirstPositionRange(clientInfo);
        } else
        {
            return result;
        }
    }

    public Map<Long, PositionRange> getBatchIdToPositionRangeMap(ClientInfo clientInfo)
    {
        String path = Zookeepers.getBatchMarkPath(clientInfo.getDestination(), clientInfo.getClientId());
        List<String> nodes = null;
        try
        {
            nodes = zooKeeperClient.getChildren(path);
        } catch (ZkNoNodeException e)
        {
            // ignore
        }

        if (CollectionUtils.isEmpty(nodes))
        {
            return Maps.newHashMap();
        }
        // 找到最大的Id
        ArrayList<Long> batchIds = new ArrayList<Long>(nodes.size());
        for (String batchIdString : nodes)
        {
            batchIds.add(Long.valueOf(batchIdString));
        }

        Collections.sort(batchIds); // 从小到大排序
        Map<Long, PositionRange> positionRanges = Maps.newLinkedHashMap();
        for (Long batchId : batchIds)
        {
            PositionRange result = getPositionRange(clientInfo, batchId);
            if (result == null)
            {// 出现为null，说明zk节点有变化，重新获取
                return getBatchIdToPositionRangeMap(clientInfo);
            } else
            {
                positionRanges.put(batchId, result);
            }
        }

        return positionRanges;
    }

    // =========== setter ==========

    public void setZooKeeperClient(ZooKeeperClient zooKeeperClient)
    {
        this.zooKeeperClient = zooKeeperClient;
    }

}
